/*
 * Barrier.cpp
 *
 *  Created on: Nov 23, 2012
 *      Author: dan
 */

#include "Barrier.h"
#include "Print.hpp"

namespace {

typedef boost::asio::ip::tcp::socket socket_t;
typedef boost::shared_ptr<socket_t> socket_ref;
typedef std::map<std::string, socket_ref> Hosts;
typedef boost::system::error_code error_code;
typedef boost::thread thread_t;
typedef boost::shared_ptr<thread_t> thread_ref;
typedef std::map<std::string, thread_ref> Threads;
typedef boost::asio::io_service ios_t;
typedef boost::shared_ptr<ios_t> ios_ref;

}

Barrier::Barrier(Network& network, Network::Team team, std::string barrier_name) :
		network(network), team(team), name(barrier_name) {
	boost::mutex::scoped_lock l(mtx);
	callbackId = network.setCallbacks(
			boost::bind(&Barrier::Event_OnConnection, this, _1, _2),
			boost::bind(&Barrier::Event_OnDisconnection, this, _1, _2),
			boost::bind(&Barrier::Event_OnMessage, this, _1, _2, _3));
	init();
}
Barrier::~Barrier() {
	network.removeCallbacks(callbackId);
}

void Barrier::Event_OnConnection(std::string id, error_code error) {
	PRINT("BARRIER("+name+"): EVENT<CONNECTION>: id="<<id<<", error="<<error);
	boost::mutex::scoped_lock l(mtx);
	if (inTeam(id) == false)
		return;
	connected[id] = true;
	state_changed.notify_one();
}
void Barrier::Event_OnDisconnection(std::string id, error_code error) {
	PRINT("BARRIER("+name+"): EVENT<DISCONNECTION>: id="<<id<<", error="<<error);
	boost::mutex::scoped_lock l(mtx);
	if (inTeam(id) == false)
		return;
	connected[id] = false;
	ready[id] = false;
	state_changed.notify_one();
}
void Barrier::Event_OnMessage(std::string id, std::string message,
		error_code error) {
	PRINT(
			"BARRIER("+name+"): EVENT<CONNECTION>: id="<<id<<", error="<<error<<", message="<<message);
	boost::mutex::scoped_lock l(mtx);
	std::string bmessage = getBarrierMessage();
	if (inTeam(id) == false || message != bmessage)
		return;
	ready[id] = true;
	state_changed.notify_one();
}

void Barrier::waitConnection() {
	boost::mutex::scoped_lock l(mtx);
	while (true) {
		if (allConnected() == true)
			return;
		state_changed.wait(l);
	}
}
void Barrier::waitConnection(boost::mutex::scoped_lock& ext_l) {
	boost::mutex::scoped_lock l(mtx);
	bool ext_l_locked = true;
	while (true) {
		if (allConnected() == true) {
			if (ext_l_locked == false)
				ext_l.lock();
			return;
		}
		ext_l_locked = false;
		ext_l.unlock();
		state_changed.wait(l);
	}
}
bool Barrier::waitOne() {
	boost::mutex::scoped_lock l(mtx);
	network.send(team, getBarrierMessage());
	while (true) {
		if (allConnected() == false) {
			PRINT("BARRIER("+name+"): allConnected = false");
			return false;
		}
		if (allReady()) {
			PRINT("BARRIER("+name+"): allReady() = true");
			return true;
		}
		state_changed.wait(l);
	}
	return false;
}

bool Barrier::waitOne(boost::mutex::scoped_lock& ext_l) {
	boost::mutex::scoped_lock l(mtx);
	network.send(team, getBarrierMessage());
	bool ext_l_locked = true;
	while (true) {
		if (allConnected() == false) {
			if (ext_l_locked == false)
				ext_l.lock();
			PRINT("BARRIER("+name+"): allConnected = false");
			return false;
		}
		if (allReady()) {
			if (ext_l_locked == false)
				ext_l.lock();
			PRINT("BARRIER("+name+"): allReady() = true");
			return true;
		}
		ext_l_locked = false;
		ext_l.unlock();
		state_changed.wait(l);
	}
	return false;
}

bool Barrier::wait() {
	boost::mutex::scoped_lock l(mtx);
	network.send(team, getBarrierMessage());
	while (true) {
		if (allConnected() == false) {
			PRINT("BARRIER("+name+"): allConnected = false");
			init();
			return false;
		}
		if (allReady()) {
			PRINT("BARRIER("+name+"): allReady() = true");
			init();
			return true;
		}
		state_changed.wait(l);
	}
	init();
	return false;
}
bool Barrier::wait(boost::mutex::scoped_lock& ext_l) {
	boost::mutex::scoped_lock l(mtx);
	network.send(team, getBarrierMessage());
	bool ext_l_locked = true;
	while (true) {
		if (allConnected() == false) {
			if (ext_l_locked == false)
				ext_l.lock();
			PRINT("BARRIER("+name+"): allConnected = false");
			init();
			return false;
		}
		if (allReady()) {
			if (ext_l_locked == false)
				ext_l.lock();
			PRINT("BARRIER("+name+"): allReady() = true");
			init();
			return true;
		}
		ext_l_locked = false;
		ext_l.unlock();
		state_changed.wait(l);
	}
	init();
	return false;
}


void Barrier::clear() {
	boost::mutex::scoped_lock l(mtx);
	init();
}

bool Barrier::allConnected() {
	for (HostList::iterator i = connected.begin(); i != connected.end(); i++)
		if (i->second == false)
			return false;
	return true;
}
bool Barrier::allReady() {
	for (HostList::iterator i = ready.begin(); i != ready.end(); i++)
		if (i->second == false)
			return false;
	return true;
}
void Barrier::init() {
	ready.clear();
	teamSet.clear();
	std::set<std::string> active = network.getConnectionsSet();
	for (size_t i = 0; i < team.size(); i++) {
		std::string member = team[i].str();
		connected[member] = active.find(member) != active.end()
				|| network.isMe(member);
		ready[member] = false || network.isMe(member);
		teamSet.insert(member);
	}
	printTeamStatus();
}
void Barrier::printTeamStatus() {
	for (size_t i = 0; i < team.size(); i++) {
		std::string member = team[i].str();
		PRINT(
				"BARRIER("+name+"): team member ["+member+"] is " <<(connected[member]?"connected":"disconnected")<<" and " <<(ready[member]?"ready":"not ready"));
	}
}
bool Barrier::inTeam(std::string id) {
	return teamSet.find(id) != teamSet.end();
}

